package com.raylu.realtime.app.dwm;

import com.alibaba.fastjson.JSON;
import com.raylu.realtime.bean.PageBean;
import com.raylu.realtime.utils.KafkaSinkUtil;
import com.raylu.realtime.utils.KafkaSourceUtil;
import com.raylu.realtime.utils.PropertiesUtil;
import org.apache.flink.api.common.functions.RichFilterFunction;
import org.apache.flink.api.common.restartstrategy.RestartStrategies;
import org.apache.flink.api.common.state.StateTtlConfig;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.runtime.state.filesystem.FsStateBackend;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.CheckpointConfig;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.Properties;

/**
 * Description:
 * <p>独立访客计算</p>
 * <p>
 * 将访客的改天的第一次访问JSON对象放入kafka主题：unique_visitor_app_group
 * </p>
 * Create by lucienoz on 2021/12/29.
 * Copyright © 2021 lucienoz. All rights reserved.
 */
public class UniqueVisitorApp {
    public static void main(String[] args) throws Exception {
        //TODO 1. 准备基本执行环境
        Properties load = PropertiesUtil.load("config.properties");
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(4);
        //TODO 1.1. 设置检查点配置
//        env.enableCheckpointing(5000L);
//        env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);
//        env.getCheckpointConfig().setCheckpointTimeout(60*1000L);
//        env.getCheckpointConfig().setCheckpointInterval(3000L);
//        env.getCheckpointConfig().enableExternalizedCheckpoints(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);
//        env.setRestartStrategy(RestartStrategies.fixedDelayRestart(3, 3000L));
        env.setStateBackend(new FsStateBackend(load.getProperty("unique.visitor.app.fsstatebackend.url")));
//        System.setProperty("HADOOP_USER_NAME", "raylu");
        //TODO 2. Flink读取Kafka DWD_PAGE_LOG主题的数据
        SingleOutputStreamOperator<PageBean> filter = env.addSource(KafkaSourceUtil.getKafkaSource(load.getProperty("base.log.app.kafka.sink-topic1"), load.getProperty("unique.visitor.app.kafka.group-id")))
                //TODO 3. 将读取的JSONSTR -> 转换成PageBean对象
                .map(r -> JSON.parseObject(r, PageBean.class))
                //TODO 4. 通过keyby将数据进行按照md或者mid进行分组
                .keyBy(r -> r.getMid())
                //TODO 5. 通过filter语义筛选出访客该天第一浏览的配置信息
                .filter(new RichFilterFunction<PageBean>() {
                    //TODO 5.1. 通过设备id进行分组
                    private ValueState<String> midDate;
                    private SimpleDateFormat dateFormat;

                    @Override
                    public void open(Configuration parameters) throws Exception {
                        super.open(parameters);
                        ValueStateDescriptor<String> stringValueStateDescriptor = new ValueStateDescriptor<>("mid-unique-date", Types.STRING);
                        stringValueStateDescriptor
                                .enableTimeToLive(StateTtlConfig.newBuilder(Time.days(1L))
                                        .neverReturnExpired()
                                        .setUpdateType(StateTtlConfig.UpdateType.OnCreateAndWrite)
                                        .build());
                        midDate = this.getRuntimeContext()
                                .getState(stringValueStateDescriptor);

                        dateFormat = new SimpleDateFormat("yyyy-MM-dd");
                    }

                    @Override
                    public boolean filter(PageBean value) throws Exception {
                        Long ts = value.ts;
                        String dateStr = dateFormat.format(new Date(ts));
                        if (midDate.value() != null && midDate.value().length() > 0 && midDate.value().equals(dateStr)) {
                            return false;
                        }
                        midDate.update(dateStr);
                        return true;
                    }
                });

        //TODO 6. 同过KafkaSink发送到kafka DWM_UNIQUE_VISITOR
        //TODO 6.1. 将PageBean转换成JSON
        filter.map(JSON::toJSONString)
                .addSink(KafkaSinkUtil.getKafkaSink(load.getProperty("unique.visitor.app.kafka.topic")));
        filter.print();


        env.execute();


    }
}
